package demo.kafka.apache;

import demo.utils.Constants;
import demo.utils.ThreadUtil;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.Before;
import org.junit.Test;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * <a href="http://cloudurable.com/blog/kafka-tutorial-kafka-consumer/index.html">Kafka 消费者</a>
 */
public class ApacheConsumerDemo {

    public static final String TOPIC = "test.stan";

    public Properties properties;

    @Before
    public void setUp() {
        properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BOOTSTRAP_SERVERS);
        // 当前消费实例所属的消费组，属于同一个组的消费实例，会负载消费消息。
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group_test_1");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 每次Poll的最大数量。注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);

        //消息的反序列化方式
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 两次Poll之间的最大允许间隔。消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，
        // 服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

    }

    /**
     * 初始化事务配置
     *
     * @param properties
     */
    static void initTransactionCfg(Properties properties) {
        // 唯一的事务ID，用于故障恢复
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transactional-id");
        // 启用幂等性是使用事务的前提
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
    }

    /**
     * 同步提交偏移量。
     * 消费者的“同步”模式通常指的是：拉取一批消息 -> 逐条或批量同步处理 -> 同步提交偏移量。
     * 在这种模式下，消费者的主线程会阻塞，等待当前批次的所有消息都被处理完毕后，才继续下一轮的拉取。
     *
     * <li>优点：简单、可靠、易于理解。能保证至少一次（At Least Once） 语义（如果处理完成后提交，提交前应用崩溃，消息会被重复消费）。
     * <li>缺点：吞吐量低。如果单条消息处理很慢，会阻塞整个消费进程。
     */
    @Test
    public void testConsumerSync() {
        // 【关键】关闭自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Arrays.asList(TOPIC));
        try {
            while (true) {
                // 1. 同步拉取消息（poll 方法本身是阻塞的，直到获取到数据或超时）
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));

                // 2. 同步处理每一条消息
                for (ConsumerRecord<String, String> record : records) {
                    // 【模拟同步处理】：这里可能是调用一个同步的数据库操作、HTTP API 等
                    System.out.printf("同步处理: topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                }

                // 3. 当前批次所有消息处理完后，同步提交偏移量
                // 这会阻塞，直到 Broker 确认提交成功或失败
                if (!records.isEmpty()) {
                    kafkaConsumer.commitSync(); // 【同步提交】
                    System.out.println("偏移量已同步提交");
                }
            }
        } finally {
            kafkaConsumer.close();
        }
    }

    /**
     * 消费者的“异步”模式指的是：拉取一批消息 -> 提交给线程池异步处理 -> 继续拉取新消息。
     * <p>
     * 处理消息和提交偏移量的操作不会阻塞主消费线程，从而极大地提高了吞吐量。
     * <p>
     * 如何实现异步消费？
     * <p>
     * 主要通过结合多线程和异步提交偏移量来实现。
     *
     * <li>优点：高吞吐量。消息处理 I/O 操作不会阻塞消息拉取。
     * <li>缺点：实现复杂。需要手动管理偏移量（可能涉及线程安全的数据结构如 ConcurrentHashMap），消息顺序无法保证，错误处理也更复杂。偏移量提交失败需要自定义重试逻辑。
     */
    @Test
    public void testConsumerAsync() {
// 【关键】关闭自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Arrays.asList(TOPIC));
        // 线程池用于异步处理消息
        ExecutorService messageProcessorThreadPool = Executors.newFixedThreadPool(5);
        try {
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                final Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new ConcurrentHashMap<>();

                for (ConsumerRecord<String, String> record : records) {
                    // 将消息提交给线程池异步处理
                    messageProcessorThreadPool.submit(() -> {
                        try {
                            // 【模拟异步处理】
                            String log = String.format("异步处理: topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
                                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
                            System.out.println(log);
                            // 记录成功处理的消息的偏移量（注意线程安全！）
                            // 通常更安全的做法是每个工作线程处理完后更新自己的偏移量映射，最后再合并
                            offsetsToCommit.put(new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1) // 提交下一条的偏移量
                            );
                        } catch (Exception e) {
                            System.err.println("处理消息失败: " + e.getMessage());
                        }
                    });
                }

                // 异步提交偏移量（不会阻塞主线程）
                if (!offsetsToCommit.isEmpty()) {
                    kafkaConsumer.commitAsync(offsetsToCommit, new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
                            if (e != null) {
                                System.err.println("异步提交失败: " + e.getMessage());
                            } else {
                                System.out.println("偏移量已异步提交: " + offsets);
                            }
                        }
                    });
                }
            }
        } finally {
            messageProcessorThreadPool.shutdown();
            kafkaConsumer.close();
        }

    }

    /**
     * 适用场景：“消费-处理-生产” 链路的精确一次（Exactly-Once） 语义。
     * <p>
     * 核心思想：将消费到的消息的偏移量（offset）也作为一条特殊消息，与生产者要发送的业务消息放在同一个 Kafka 事务中提交。
     * 这样就能保证：业务消息成功发送和偏移量成功提交是原子性的。
     * <p>
     * 如何配置和使用？
     * <p>
     * <li>生产者配置：和普通事务生产者一样，需要 transactional.id 和 enable.idempotence。
     * <li>消费者配置：
     * isolation.level = read_committed：必须配置。消费者只会读取已提交事务的消息，避免读到生产者事务中止的脏数据。
     * enable.auto.commit = false：必须关闭自动提交，偏移量将由事务同步提交。
     */
    @Test
    public void testTransaction() {
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 关闭自动提交
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); // 【关键】只读已提交的消息

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Arrays.asList(TOPIC));

        // 生产者配置
        Properties producerProperties = ApacheProducerDemo.initProducerProperty();
        ApacheProducerDemo.initTransactionCfg(producerProperties);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProperties);
        producer.initTransactions();// 初始化事务

        try {
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                if (records.isEmpty()) {
                    continue;
                }
                try {
                    // 开始一个事务
                    producer.beginTransaction();
                    // 处理消息并发送到输出Topic
                    for (ConsumerRecord<String, String> record : records) {
                        // 【模拟异步处理】
                        String log = String.format("异步处理: topic = %s, partition = %d, offset = %d, key = %s, value = %s\n",
                                record.topic(), record.partition(), record.offset(), record.key(), record.value());
                        System.out.println(log);
                        producer.send(new ProducerRecord<>("topic_output", record.key(), record.value()));
                    }

                    // 【核心】计算本次消费的偏移量，并将其作为事务的一部分发送到一个特殊的内部Topic
                    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        offsetsToCommit.put(partition, new OffsetAndMetadata(lastOffset + 1));
                    }

                    // 将偏移量发送到事务中
                    /**
                     * 在Kafka中，消费者的偏移量本质上也是存储在Kafka内部的一个特殊主题（__consumer_offsets）中的消息。
                     *
                     * 普通主题（如 user-events）：存储你的业务数据。
                     * 内部主题（__consumer_offsets）：存储各个消费者组的消费进度（即offset）。
                     * 所以，当你调用 consumer.commitSync() 时，底层的操作其实就是消费者客户端向 __consumer_offsets 主题发送一条消息，
                     * 内容大概是：“组A在分区0的消费进度现在是105了”。
                     * <p>
                     * 输出主题的业务消息 和 提交到 __consumer_offsets 的进度消息 要么都成功，要么都失败。这就是一个典型的分布式事务问题。
                     * sendOffsetsToTransaction 就是这个解决方案的关键。它做了两件事：
                     * <li>它告诉事务协调器（Transaction Coordinator）：“嘿，等会儿你提交事务时，除了我之前用 producer.send() 发的那些业务消息，还得帮我把这批偏移量消息也一起提交了”。
                     * <li>它将这些偏移量信息暂存起来，等待事务最终提交。
                     */
                    producer.sendOffsetsToTransaction(offsetsToCommit, kafkaConsumer.groupMetadata());

                    // 提交事务（包括业务消息和消费偏移量）
                    producer.commitTransaction();
                    System.out.println("事务提交成功");

                } catch (Exception e) {
                    // 中止事务，所有操作（发送的消息和偏移量）都会回滚
                    producer.abortTransaction();
                    System.err.println("事务失败，已回滚: " + e.getMessage());
                    // 可以根据异常类型决定是否重试本轮消息
                }
            }
        } finally {
            producer.close();
            kafkaConsumer.close();
        }

    }

    /**
     * 消费者
     */
    @Test
    public void testConsumerSubscribe() {

        //构造消息对象，也即生成一个消费实例。
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        //设置消费组订阅的Topic，可以订阅多个。如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
        kafkaConsumer.subscribe(Arrays.asList(TOPIC), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                // 将偏移设置到最开始
                kafkaConsumer.seekToBeginning(collection);
            }
        });
        long timeout = 1000;
        //循环消费消息。
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(timeout);
            System.err.println("获取消息数量：" + records.count());
            //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
            // 建议开一个单独的线程池来消费消息，然后异步返回结果。
            for (ConsumerRecord<String, String> record : records) {
                System.err.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                        record.value());
                System.err.println("");
                System.err.println("----");
            }
            ThreadUtil.safeSleep(2000);
        }

    }


    /**
     * assign() 是 Kafka Consumer API 中的一个方法，它允许消费者直接指定要消费的特定主题分区（Topic-Partition），
     * 而不是让 Kafka 的消费者组协调器（Group Coordinator）自动进行分区分配。
     *
     * Assign特性：
     * <li>分区分配: 手动、静态。由应用程序代码显式指定。
     * <li>消费者组:	不使用消费者组机制。group.id 配置无效或仅用于偏移量管理。
     * <li>负载均衡:	无自动负载均衡。如果新增或减少消费者，必须手动重新分配分区。
     * <li>容错性	较低。如果一个消费者实例宕机，它负责的分区将无人消费，除非你手动启动另一个消费者来接管
     * <li>使用场景:	特殊场景，如定点消费、回溯数据、独立消费。
     *
     */
    @Test
    public void testConsumerAssign() {
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            // 创建一个TopicPartition列表，指定要消费的确切分区
            List<TopicPartition> partitionsToAssign = Arrays.asList(
                    new TopicPartition("my-topic", 0), // 消费 my-topic 的 0 号分区
                    new TopicPartition("my-topic", 2), // 消费 my-topic 的 2 号分区
                    new TopicPartition("another-topic", 1) // 消费 another-topic 的 1 号分区
            );
            // 将分区直接分配给消费者
            consumer.assign(partitionsToAssign);
            long timeout = 0;
            for (int i = 0; i < 10; i++) {
                ConsumerRecords<String, String> records = consumer.poll(timeout);
                System.err.println("获取消息数量：" + records.count());
                for (ConsumerRecord<String, String> record : records) {
                    System.err.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                            record.value());
                }
                ThreadUtil.safeSleep(2000);
            }
        }

    }

}
